血缘|表里_SparkSQL字段血缘扩展实践！

篇首语：本文由编程笔记#小编为大家整理，主要介绍了Spark SQL字段血缘扩展实践！相关的知识，希望对你有一定的参考价值。

目录

• 一、背景
• 二、前期调研
• 三、Spark SQL 扩展
• • 3.1 Spark 可扩展的内容
  • 3.2 实现自己的扩展
  • 3.3 扩展的规则类
  • 3.4 具体的实现方法
  
  
• 四、总结

字段血缘是在表处理的过程中将字段的处理过程保留下来。为什么会需要字段血缘呢&＃xff1f;

有了字段间的血缘关系&＃xff0c;便可以知道数据的来源去处&＃xff0c;以及字段之间的转换关系&＃xff0c;这样对数据的质量&＃xff0c;治理有很大的帮助。

Spark SQL 相对于 Hive 来说通常情况下效率会比较高&＃xff0c;对于运行时间、资源的使用上面等都会有较大的收益。

平台计划将 Hive 任务迁移到 Spark SQL 上&＃xff0c;同时也需要实现字段血缘的功能。

开发前我们做了很多相关调研&＃xff0c;从中得知 Spark 是支持扩展的&＃xff1a;允许用户对 Spark SQL 的 SQL 解析、逻辑计划的分析和检查、逻辑计划的优化、物理计划的形成等进行扩展。

该方案可行&＃xff0c;且对 Spark 的源码没有改动&＃xff0c;代价也比较小&＃xff0c;确定使用该方案。

3.1 Spark 可扩展的内容

SparkSessionExtensions是比较重要的一个类&＃xff0c;其中定义了注入规则的方法&＃xff0c;现在支持以下内容&＃xff1a;

【Analyzer Rules】逻辑计划分析规则

【Check Analysis Rules】逻辑计划检查规则

【Optimizer Rules.】 逻辑计划优化规则

【Planning Strategies】形成物理计划的策略

【Customized Parser】自定义的sql解析器

【(External) Catalog listeners catalog】监听器

在以上六种可以用户自定义的地方&＃xff0c;我们选择了【Check Analysis Rules】。因为该检查规则在方法调用的时候是不需要有返回值的&＃xff0c;也就意味着不需要对当前遍历的逻辑计划树进行修改&＃xff0c;这正是我们需要的。

而【Analyzer Rules】、【Optimizer Rules】则需要对当前的逻辑计划进行修改&＃xff0c;使得我们难以迭代整个树&＃xff0c;难以得到我们想要的结果。

3.2 实现自己的扩展

class ExtralSparkExtension extends (SparkSessionExtensions &＃61;> Unit)
override def apply(spark: SparkSessionExtensions): Unit &＃61;
//字段血缘
spark.injectCheckRule(FieldLineageCheckRuleV3)
//sql解析器
spark.injectParser case (_, parser) &＃61;> new ExtraSparkParser(parser)




上面按照这种方式实现扩展&＃xff0c;并在 apply 方法中把自己需要的规则注入到 SparkSessionExtensions 即可&＃xff0c;除了以上四种可以注入的以外还有其他的规则。要让 ExtralSparkExtension 起到作用的话我们需要在spark-default.conf下配置

spark.sql.extensions&＃61;org.apache.spark.sql.hive.ExtralSparkExtension

在启动 Spark 任务的时候即可生效。

注意到我们也实现了一个自定义的SQL解析器&＃xff0c;其实该解析器并没有做太多的事情。只是在判断如果该语句包含insert的时候就将 SQLText&＃xff08;SQL语句&＃xff09;设置到一个为FIELD_LINE_AGE_SQL&＃xff0c;之所以将SQLText放到FIELD_LINE_AGE_SQL里面。因为在 DheckRule 里面是拿不到SparkPlan的我们需要对SQL再次解析拿到 SprkPlan&＃xff0c;而FieldLineageCheckRuleV3的实现也特别简单&＃xff0c;重要的在另一个线程实现里面。

这里我们只关注了insert语句&＃xff0c;因为插入语句里面有从某些个表里面输入然后写入到某个表。

class ExtraSparkParser(delegate: ParserInterface) extends ParserInterface with Logging
override def parsePlan(sqlText: String): LogicalPlan &＃61;
val lineAgeEnabled &＃61; SparkSession.getActiveSession
.get.conf.getOption("spark.sql.xxx-xxx-xxx.enable").getOrElse("false").toBoolean
logDebug(s"SqlText: $sqlText")
if(sqlText.toLowerCase().contains("insert"))
if(lineAgeEnabled)
if(FIELD_LINE_AGE_SQL_COULD_SET.get())
//线程本地变量在这里
FIELD_LINE_AGE_SQL.set(sqlText)

FIELD_LINE_AGE_SQL_COULD_SET.remove()


delegate.parsePlan(sqlText)

//调用原始的sqlparser
override def parseExpression(sqlText: String): Expression &＃61;
delegate.parseExpression(sqlText)

//调用原始的sqlparser
override def parseTableIdentifier(sqlText: String): TableIdentifier &＃61;
delegate.parseTableIdentifier(sqlText)

//调用原始的sqlparser
override def parseFunctionIdentifier(sqlText: String): FunctionIdentifier &＃61;
delegate.parseFunctionIdentifier(sqlText)

//调用原始的sqlparser
override def parseTableSchema(sqlText: String): StructType &＃61;
delegate.parseTableSchema(sqlText)

//调用原始的sqlparser
override def parseDataType(sqlText: String): DataType &＃61;
delegate.parseDataType(sqlText)




3.3 扩展的规则类

case class FieldLineageCheckRuleV3(sparkSession:SparkSession) extends (LogicalPlan&＃61;>Unit )
val executor: ThreadPoolExecutor &＃61;
ThreadUtils.newDaemonCachedThreadPool("spark-field-line-age-collector",3,6)
override def apply(plan: LogicalPlan): Unit &＃61;
val sql &＃61; FIELD_LINE_AGE_SQL.get
FIELD_LINE_AGE_SQL.remove()
if(sql !&＃61; null)
//这里我们拿到sql然后启动一个线程做剩余的解析任务
val task &＃61; new FieldLineageRunnableV3(sparkSession,sql)
executor.execute(task)





很简单&＃xff0c;我们只是拿到了 SQL 然后便启动了一个线程去得到 SparkPlan&＃xff0c;实际逻辑在

FieldLineageRunnableV3。

3.4 具体的实现方法

3.4.1 得到 SparkPlan

我们在 run 方法中得到 SparkPlan&＃xff1a;

override def run(): Unit &＃61;
val parser &＃61; sparkSession.sessionState.sqlParser
val analyzer &＃61; sparkSession.sessionState.analyzer
val optimizer &＃61; sparkSession.sessionState.optimizer
val planner &＃61; sparkSession.sessionState.planner
............
val newPlan &＃61; parser.parsePlan(sql)
PASS_TABLE_AUTH.set(true)
val analyzedPlan &＃61; analyzer.executeAndCheck(newPlan)
val optimizerPlan &＃61; optimizer.execute(analyzedPlan)
//得到sparkPlan
val sparkPlan &＃61; planner.plan(optimizerPlan).next()
...............
if(targetTable !&＃61; null)
val levelProject &＃61; new ArrayBuffer[ArrayBuffer[NameExpressionHolder]]()
val predicates &＃61; new ArrayBuffer[(String,ArrayBuffer[NameExpressionHolder])]()
//projection
projectionLineAge(levelProject, sparkPlan.child)
//predication
predicationLineAge(predicates, sparkPlan.child)
...............


为什么要使用 SparkPlan 呢&＃xff1f;当初我们考虑的时候&＃xff0c;物理计划拿取字段关系的时候是比较准的&＃xff0c;且链路比较短也更直接。

在这里补充一下 Spark SQL 解析的过程如下&＃xff1a;

经过SqlParser后会得到逻辑计划&＃xff0c;此时表名、函数等都没有解析&＃xff0c;还不能执行&＃xff1b;经过Analyzer会分析一些绑定信息&＃xff0c;例如表验证、字段信息、函数信息&＃xff1b;经过Optimizer 后逻辑计划会根据既定规则被优化&＃xff0c;这里的规则是RBO&＃xff0c;当然 Spark 还支持CBO的优化&＃xff1b;经过SparkPlanner后就成了可执行的物理计划。

我们看一个逻辑计划与物理计划对比的例子&＃xff1a;
一个 SQL 语句&＃xff1a;


select item_id,TYPE,v_value,imei from t1
union all
select item_id,TYPE,v_value,imei from t2
union all
select item_id,TYPE,v_value,imei from t3


逻辑计划是这样的&＃xff1a;

物理计划是这样的&＃xff1a;

显然简化了很多。
得到 SparkPlan 后&＃xff0c;我们就可以根据不同的SparkPlan节点做迭代处理。
我们将字段血缘分为两种类型&＃xff1a;projection&＃xff08;select查询字段&＃xff09;、predication(wehre查询条件)。
这两种是一种点对点的关系,即从原始表的字段生成目标表的字段的对应关系。
想象一个查询是一棵树&＃xff0c;那么迭代关系会如下从树的顶端开始迭代&＃xff0c;直到树的叶子节点&＃xff0c;叶子节点即为原始表&＃xff1a;

那么我们迭代查询的结果应该为

id ->tab1.id ,

name->tab1.name,tabb2.name,

age→tabb2.age。

注意到有该变量

val levelProject &＃61; new ArrayBuffer

ArrayBuffer[NameExpressionHolder]&＃xff0c;通过projecti-onLineAge 迭代后 levelProject 存储了顶层id&＃xff0c;name,age对应的&＃xff08;tab1.id&＃xff09;&＃xff0c;&＃xff08;tab1.name,tabb2.name&＃xff09;&＃xff0c;&＃xff08;tabb2.age&＃xff09;。

当然也不是简单的递归迭代&＃xff0c;还需要考虑特殊情况例如&＃xff1a;Join、ExplandExec、Aggregate、Explode、GenerateExec等都需要特殊考虑。
例子及效果&＃xff1a;

SQL:

with A as (select id,name,age from tab1 where id > 100 ) ,
C as (select id,name,max(age) from A group by A.id,A.name) ,
B as (select id,name,age from tabb2 where age > 28)
insert into tab3
select C.id,concat(C.name,B.name) as name, B.age from
B,C where C.id &＃61; B.id


效果&＃xff1a;


"edges": [

"sources": [
3
],
"targets": [
0
],
"expression": "id",
"edgeType": "PROJECTION"
,

"sources": [
4,
7
],
"targets": [
1
],
"expression": "name",
"edgeType": "PROJECTION"
,

"sources": [
5
],
"targets": [
2
],
"expression": "age",
"edgeType": "PROJECTION"
,

"sources": [
6,
3
],
"targets": [
0,
1,
2
],
"expression": "INNER",
"edgeType": "PREDICATE"
,

"sources": [
6,
5
],
"targets": [
0,
1,
2
],
"expression": "((((default.tabb2.&＃96;age&＃96; IS NOT NULL) AND (CAST(default.tabb2.&＃96;age&＃96; AS INT) > 28)) AND (B.&＃96;id&＃96; > 100)) AND (B.&＃96;id&＃96; IS NOT NULL))",
"edgeType": "PREDICATE"
,

"sources": [
3
],
"targets": [
0,
1,
2
],
"expression": "((default.tab1.&＃96;id&＃96; IS NOT NULL) AND (default.tab1.&＃96;id&＃96; > 100))",
"edgeType": "PREDICATE"

],
"vertices": [

"id": 0,
"vertexType": "COLUMN",
"vertexId": "default.tab3.id"
,

"id": 1,
"vertexType": "COLUMN",
"vertexId": "default.tab3.name"
,

"id": 2,
"vertexType": "COLUMN",
"vertexId": "default.tab3.age"
,

"id": 3,
"vertexType": "COLUMN",
"vertexId": "default.tab1.id"
,

"id": 4,
"vertexType": "COLUMN",
"vertexId": "default.tab1.name"
,

"id": 5,
"vertexType": "COLUMN",
"vertexId": "default.tabb2.age"
,

"id": 6,
"vertexType": "COLUMN",
"vertexId": "default.tabb2.id"
,

"id": 7,
"vertexType": "COLUMN",
"vertexId": "default.tabb2.name"

]

四、总结

在 Spark SQL 的字段血缘实现中&＃xff0c;我们通过其自扩展&＃xff0c;首先拿到了 insert 语句&＃xff0c;在我们自己的检查规则中拿到

SQL 语句&＃xff0c;通过SparkSqlParser、Analyzer、Optimizer、SparkPlanner&＃xff0c;最终得到了物理计划。

我们通过迭代物理计划&＃xff0c;根据不同执行计划做对应的转换&＃xff0c;然后就得到了字段之间的对应关系。当前的实现是比较简单的&＃xff0c;字段之间是直线的对应关系&＃xff0c;中间过程被忽略&＃xff0c;如果想实现字段的转换的整个过程也是没有问题的。